//package com.ruoyi.common.kafka;
//
//import org.apache.kafka.clients.producer.KafkaProducer;
//import org.apache.kafka.clients.producer.ProducerConfig;
//import org.apache.kafka.clients.producer.ProducerRecord;
//import org.apache.kafka.clients.producer.RecordMetadata;
//
//import java.util.Properties;
//import java.util.concurrent.ExecutionException;
//import java.util.concurrent.Future;
//
///**
// * kafka生产者
// * 创建kafka生产者并生产消息的步骤：
// * 1.启动zookeeper和kafka
// * 2.创建topic
// * 3.启动producer
// *
// * @Author: 倚天照海
// */
//public class MyProducer {
//    /**
//     * 1.创建topic：进入到kafka安装目录的bin目录下，执行kafka-topics.sh(Linux系统)或windows\kafka-topics.bat(windows系统)脚本
//     * Linux系统: bin/kafka-topics.sh --zookeeper 192.168.10.188:2181/kafka --create --replication-factor 2 --partitions 2 --topic testKafka
//     * windows系统: bin\windows\kafka-topics.bat --zookeeper 192.168.10.188:2181 --create --replication-factor 2 --partitions 2 --topic testKafka
//     * 2.启动producer
//     * Linux系统: bin/kafka-console-producer.sh --broker-list 192.168.10.188:9092 --topic testKafka
//     * windows系统: bin\windows\kafka-console-producer.bat --broker-list 192.168.10.188:9092 --topic testKafka
//     */
//    public void produce() throws ExecutionException, InterruptedException {
//        MyKafkaConfig kafkaConfig = new MyKafkaConfig();
//        Properties properties = new Properties();
//        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConfig.getBootstrapServer());
//        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, kafkaConfig.getKeySerializerClass());
//        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, kafkaConfig.getValueSerializerClass());
//        String topic = kafkaConfig.getTopic();
//
//        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
//        doProduce(producer, topic);
//    }
//
//    private void doProduce(KafkaProducer<String, String> producer, String topic) throws ExecutionException, InterruptedException {
//        while (true) {
//            for (int i = 0; i < 3; i++) {
//                for (int j = 0; j < 3; j++) {
//                    ProducerRecord<String, String> record = new ProducerRecord<>(topic, "item" + j, "price" + i);
//                    Future<RecordMetadata> future = producer.send(record);
//                    RecordMetadata recordMetadata = future.get();
//                    int partition = recordMetadata.partition();
//                    long offset = recordMetadata.offset();
//                    System.out.println("key=" + record.key() + ", value=" + record.value()
//                            + ", partition=" + partition + ", offset=" + offset);
//                }
//            }
//        }
//    }
//}
